package main

import (
	"github.com/segmentio/kafka-go"
	"log"
	"time"
)

func writeByConn() {
	//topic := "my-topic"
	//partition := 0
	conn, err := kafka.Dial("tcp", "192.168.149.131:9092")
	//conn, err := kafka.DialLeader(context.Background(), "tcp", "192.168.149.131:9092", topic, partition)
	if err != nil {
		log.Fatal("faild to dial leader:", err)
	} else {
		log.Println("dial success\n")
	}
	err = conn.CreateTopics(kafka.TopicConfig{
		Topic:             "topic2",
		NumPartitions:     1,
		ReplicationFactor: 1,
	})
	if err != nil {
		log.Fatal("faild to create topic:", err)
	} else {
		log.Println("create topic success\n")
	}

	//设置发送超时
	conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
	conn.Write([]byte("hello"))
	//发送消息
	_, err = conn.WriteMessages(
		kafka.Message{
			Topic:     "topic1",
			Partition: 0,
			Value:     []byte("one")},
		kafka.Message{
			Topic:     "topic1",
			Partition: 0,
			Value:     []byte("two")},
		kafka.Message{
			Topic:     "topic1",
			Partition: 0,
			Value:     []byte("three")},
	)
	if err != nil {
		log.Fatal("failed to write message:", err)
	}

	//关闭连接
	if err := conn.Close(); err != nil {
		log.Fatalln("faild to close writer", err)
	}

}

func main() {
	writeByConn()
}
